package cn.wangjie.spark.mysql

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SparkWriteMySQL {

  def main(args: Array[String]): Unit = {
    // 构建SparkContext上下文实例对象
		val sc: SparkContext = {
			// 1.a 创建SparkConf对象，设置应用属性，比如应用名称和运行模式
			val sparkConf = new SparkConf()
    			.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
    			.setMaster("local[2]")
			// 1.b 创建实例对象
			SparkContext.getOrCreate(sparkConf)
		}
		
		// 1. 从本地文件系统读取数据
		val inputRDD: RDD[String] = sc.textFile("datas/wordcount/wordcount.data")
		
		// 2. 分析数据，调用RDD中函数
		val resultRDD: RDD[(String, Int)] = inputRDD
			// 过滤不合格的数据
            .filter(line => null != line && line.trim.length > 0)
			// 每行数据分割为单词
			.flatMap(line => line.trim.split("\\s+"))
			// 每个单词出现一次
			.mapPartitions{iter => iter.map(word => (word, 1))}
			// 分组，聚合统计
			.reduceByKey((tmp, item) => tmp + item)
		
		// TODO: 输出结果数据至MySQL表中：tb_wordcount
		resultRDD
			// 降低分区数据
			.coalesce(1)
			// 针对每个分区数据进行操作
            .foreachPartition{iter => saveToMySQL(iter)}
		
		// 应用结束，关闭资源
		sc.stop()
	}
	
	
	def saveToMySQL(iter: Iterator[(String, Int)]): Unit = {
		var conn: Connection = null
		var pstmt: PreparedStatement = null
		
		try{
			// step1. 获取连接
			// 加载驱动类
			Class.forName("com.mysql.cj.jdbc.Driver")
			// 获取连接
			conn = DriverManager.getConnection(
				"jdbc:mysql://node1.itcast.cn:3306/?serverTimezone=UTC&characterEncoding=utf8&useUnicode=true", //
				"root", //
				"123456"
			)
			
			val insertSQL = "INSERT INTO db_test.tb_wordcount (word, count) VALUES (?, ?)"
			pstmt = conn.prepareStatement(insertSQL)
			
			// TODO: 设置手动提交
			conn.setAutoCommit(false)
			
			// step2. 批量写入数据至表中
			iter.foreach{case (word, count) =>
				pstmt.setString(1, word)
				pstmt.setString(2, count.toString)
				// 加入批次
				pstmt.addBatch()
			}
			
			// 批量插入
			pstmt.executeBatch()
			
			// TODO: 手动提交
			conn.commit()
		}catch {
			case e: Exception => e.printStackTrace()
		}finally {
			// step3. 关闭连接
			if(null != pstmt) pstmt.close()
			if(null != conn) conn.close()
		}
		
	}
	
}
